Java8 增强的Future:CompletableFuture(笔记) - Darcy_wang - 博客园

创建时间：2017/12/12 10:27
来源：http://www.cnblogs.com/ten951/p/6590604.html


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
     CompletableFuture是Java8新增的一个超大型工具类,为什么说她大呢?因为一方面它实现了Future接口,更重要的是,它实现了CompletionStage接口.这个接口也是Java8新增加的,而CompletionStage拥有多达约40种方法,
 
    * 完成了通知我
    * 异步执行任务
 
通过CompletableFuture提供进一步封装,我们很容易实现Future模式那样的异步调用,例如:
public static Integer cale(Integer para) {
    try {
        Thread.sleep(1000);
 
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return para * para;
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
 
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> cale(50));
    System.out.println(future.get());
}
 
上述代码中CompletableFuture.supplyAsync()方法构造了一个CompletableFuture实例,在supplyAsync()函数中,他会在一个新的线程中,执行传入的参数.在这里,,他会执行calc()方法,而calc()方法的执行可能是比较慢的,但是不影响CompletableFuture实例的构造速度,因此supplyAsync()会立即返回,他返回的CompletableFuture对象实例就可以作为这次调用的契约,在将来任何场合,用于获得最终的计算结果.
在CompletableFuture中,类似的工厂方法有以下几个:
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
                                                   Executor executor)
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable,
                                               Executor executor)
 
     其中supplyAsync()方法用于那些需要返回值的场景,比如计算某个数据,而runAsync()方法用于没有返回值的场景,比如,仅仅是简单地执行某一个异步动作.
     在这两对方法中,都有一个方法可以接手一个Executor参数,这使我们可以让Supplier<U>或者Runnable在指定的线程池工作,如果不指定,则在默认的系统公共的ForkJoinPool.common线程池中执行.
 
    * 流式调用
 
在前文中我已经简单的提到,CompletionStage的约40个接口为函数式编程做准备的,在这里,就让我们看一下,如果使用这些接口进行函数式的流式API调用:
CompletableFuture<Void> future = CompletableFuture
        .supplyAsync(() -> cale(50))
        .thenApply(i -> Integer.toString(i))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
future.get();
 
     上述代码中,使用supplyAsync()函数执行了一个异步任务,接着连续使用流式调用对任务处理结果进行在加工,直到最后的结果输出:
 
    * CompletableFuture中的异常处理
 
CompletableFuture<Void> future = CompletableFuture
        .supplyAsync(() -> cale(50))
        .exceptionally(ex -> {
            System.out.println("ex.toString() = " + ex.toString());
            return 0;
        })
        .thenApply(i -> Integer.toString(i))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
future.get();
 
 
    * 组合多个CompletableFuture
 
CompletableFuture还允许你将多个CompletableFuture进行组合,一种方法是使用thenCompose(),它的签名如下:
public <U> CompletableFuture<U> thenCompose(
    Function<? super T, ? extends CompletionStage<U>> fn)
 
一个CompletableFuture可以执行完成后,将执行结果通过Function传递给下一个CompletionStage进行处理:
 
CompletableFuture<Void> future = CompletableFuture
        .supplyAsync(() -> cale(50))
        .thenCompose(i -> CompletableFuture
                .supplyAsync(() -> cale(i)))
        .thenApply(i -> Integer.toString(i))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
future.get();
 
 
另外一种组和多个CompletableFuture的方法是thenCombine(),它的签名如下:
public <U,V> CompletableFuture<V> thenCombine(
    CompletionStage<? extends U> other,
    BiFunction<? super T,? super U,? extends V> fn)
 
     方法thenCombine()首先完成当前CompletableFuture和other的执行,接着,将这两者的执行结果传递给BiFunction(该接口接受两个参数,并有一个返回值),并返回代表BiFuntion实例的CompletableFuture对象:
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> cale(50));
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> cale(25));
 
CompletableFuture<Void> fu = future1.thenCombine(future2, (i, j) -> (i + j))
        .thenApply(str -> "\"" + str + "\"")
        .thenAccept(System.out::println);
fu.get();